package com.shujia.streeam

import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.ReceiverInputDStream

object Demo2UpdataStateByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("Demo1WordCount")

    val sc = new SparkContext(conf)

    val initRDD = sc.parallelize(List(("shujia", 100), ("java", 10000)))


    /**
      *
      * 创建 streamming上下文对象  需要指定间隔时间   延迟时间
      *
      */
    val ssc = new StreamingContext(sc, Durations.seconds(5))


    //设置checkpoint路径 用于存储之前batch的状态
    ssc.checkpoint("spark/data/ssccheckpoint")


    /**
      * 连接socket端口创建Dstream
      *
      * 在 linux  中执行  nc  -lk 8888   绑定一个端口提供数据服务
      *
      * yum install nc
      *
      */
    val ds: ReceiverInputDStream[String] = ssc.socketTextStream("node1", 8888, StorageLevel.MEMORY_AND_DISK_SER)

    /**
      * updateStateByKey   更新key的状态   有状态算子
      *
      */

    /**
      * seq  当前batch  一个key的所有数据
      * state  之前所有batch的状态（统计结果）    之前batch的状态需要保存到checkpoint中
      */
    val fun = (seq: Seq[Int], state: Option[Int]) => {

      //1、统计当前batch
      val count = seq.sum


      //之前的状态
      val last = state.getOrElse(0)


      //将当前bacth统计结果加上之前batch统计结果  返回
      Option(count + last)

    }


    /**
      * updateStateByKey  每隔5秒执行一次  每次执行的时候更新之前batch key的状态
      *
      */
    ds.flatMap(line => line.split(","))
      .map(word => (word, 1))
      //.updateStateByKey(fun) //更新每一个key的状态
      .updateStateByKey(fun, new HashPartitioner(2), initRDD) //initRDD  key初始值
      .print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}
